package com.fw.kafka_chapter_1.transaction;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @Author:yanwei
 * @Date: 2020/9/5 - 23:23
 * 开始事务的demo小样
 */
@Slf4j
public class ProducerTransaction {
    private static final String TOPIC = "transaction_fw";
    private static final String BROKER_LIST = "192.168.227.129:9092";
    //客户端提供的唯一ID
    private static final String TRANSACTION_ID = "transaction_fw_id";
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,BROKER_LIST);
        // key 序列化器指定
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // value 序列化执行
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
       // 幂等性显示化
        properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true);
        properties.put(ProducerConfig.ACKS_CONFIG,"all");
        // 指定事务ID
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,TRANSACTION_ID);
        KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);

        // 初始化事务
        producer.initTransactions();
        // 开启事务
        producer.beginTransaction();
        try {
            ProducerRecord<String,String> producerRecord1 = new ProducerRecord(TOPIC,"key","fw-1");
            producer.send(producerRecord1);
            int i  =  1 / 0;
            ProducerRecord<String,String> producerRecord2 = new ProducerRecord(TOPIC,"key","fw-1");
            producer.send(producerRecord2);
            ProducerRecord<String,String> producerRecord3 = new ProducerRecord(TOPIC,"key","fw-1");
            producer.send(producerRecord3);
            // 提交事务
            producer.commitTransaction();
        }catch (Exception e){
            // 回滚事务
            producer.abortTransaction();
        }
        producer.close();


    }
}
